/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package dk.itu.samd.group2.tagapastry;

import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.p2p.commonapi.CancellableTask;
import rice.p2p.commonapi.Endpoint;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.Node;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.commonapi.RouteMessage;
import rice.p2p.commonapi.rawserialization.InputBuffer;
import rice.p2p.commonapi.rawserialization.MessageDeserializer;
import rice.pastry.commonapi.PastryIdFactory;

import java.io.IOException;
import java.sql.Array;
import java.util.Date;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Vector;

import dk.itu.samd.group2.tagapastry.messages.IndexExpansionConfirmationMsg;
import dk.itu.samd.group2.tagapastry.messages.IndexExpansionMsg;
import dk.itu.samd.group2.tagapastry.messages.IndexRetractionRequestMsg;
import dk.itu.samd.group2.tagapastry.messages.IndexRetractionResponseMsg;
import dk.itu.samd.group2.tagapastry.messages.MaintenanceMsg;
import dk.itu.samd.group2.tagapastry.messages.ReplicationMaintenanceMsg;
import dk.itu.samd.group2.tagapastry.messages.ReplicationMsg;
import dk.itu.samd.group2.tagapastry.messages.ResultsMsg;
import dk.itu.samd.group2.tagapastry.messages.SearchMsg;
import dk.itu.samd.group2.tagapastry.messages.TagMsg;
import dk.itu.samd.group2.tagapastry.messages.TagapastryMsg;
import dk.itu.samd.group2.tagapastry.messages.UntagMsg;
import dk.itu.samd.group2.tagapastry.messages.UpdateSizeMsg;
import dk.itu.samd.group2.tagapastry.util.NodeStatistics;

/**
 *
 * @author frgr
 */
public class TagapastryImpl implements Tagapastry, Iterable<Index> {

	protected Hashtable<Tag, Index> indexes;
	protected Endpoint endpoint;
	protected Node node;
	protected String instance;
	protected Environment environment;
	protected CancellableTask maintenanceTask, replicationMntTask;
	protected Logger logger;
	protected Hashtable<Tag, Vector<IndexEntry>> results;
	protected TagapastryClient client;
	private Hashtable<NodeHandle, NodeState> leafSetData;
	private int retThreshold, expThreshold;
	private TaggingBlackBox taggingBlackBox;
	private boolean shouldCatchData;
	private NodeStatistics stats;
	
	public TagapastryImpl(Node node, String clientName, TagapastryClient client) {

		// Setting a name for the app to avoid ambiguity between several 
		// tagapastry apps running on the same node
		this.instance = clientName + "_tagapastry";

		// We are only going to use one instance of this application on each PastryNode
		this.endpoint = node.buildEndpoint(this, this.instance);
		this.node = node;
		this.client = client;
		this.environment = node.getEnvironment();
		this.logger = this.environment.getLogManager().getLogger(this.getClass(), this.instance);
		setLoggingLevel(Logger.OFF);
		TagFactory.setIdFactory(new PastryIdFactory(environment));
		// hashtable capacity is set to a low number because
		// it is unlikely that there will be too many indexes
		// in comparison to the number of peers in the system
		this.indexes = new Hashtable<Tag, Index>(5);
		
		this.leafSetData = new Hashtable<NodeHandle, NodeState>(10);

		taggingBlackBox = new TaggingBlackBox(true);
		
		// No logging by default
		shouldCatchData = false; 
		
		// TODO as Fred suggested, invoke this routine procedure only 
		// when the node has at least 1 child index
		// this.maintenanceTask = endpoint.scheduleMessage(new MaintenanceMsg(this.getNodeHandle()), 4000, 4000);
				
		this.results = new Hashtable<Tag, Vector<IndexEntry>>(10);

		this.endpoint.setDeserializer(new MessageDeserializer() {

			public Message deserialize(InputBuffer buf, short type, int priority,
					NodeHandle sender) throws IOException {

				try {
					switch (type) {
					case TagMsg.TYPE:
						return new TagMsg(buf, endpoint);
					case SearchMsg.TYPE:
						return new SearchMsg(buf, endpoint);
					case ResultsMsg.TYPE:
						return new ResultsMsg(buf, endpoint);
					case UntagMsg.TYPE:
						return new UntagMsg(buf, endpoint);
					case IndexExpansionMsg.TYPE:
						return new IndexExpansionMsg(buf, endpoint);
					case IndexRetractionRequestMsg.TYPE:
						return new IndexRetractionRequestMsg(buf, endpoint);
					case IndexExpansionConfirmationMsg.TYPE:
						return new IndexExpansionConfirmationMsg(buf, endpoint);
					case IndexRetractionResponseMsg.TYPE:
						return new IndexRetractionResponseMsg(buf, endpoint);
					case UpdateSizeMsg.TYPE:
						return new UpdateSizeMsg(buf, endpoint);
					case ReplicationMsg.TYPE:
						return new ReplicationMsg(buf,endpoint);
					}
				} catch (IOException e) {
					if (logger.level <= Logger.SEVERE) 
						logger.log("Exception in deserializer in " + endpoint + ":" + instance + " " + endpoint.getDeserializer());
					throw e;
				}
				throw new IllegalArgumentException("Unknown type: " + type); 
			}    
		});

		// now we can receive messages
		this.endpoint.register();
	}
	
	public void setLoggingLevel(int level) {
		this.logger.level = level;
	}

	@Override
	public void deliver(Id id, Message msg) {
		TagapastryMsg tagaMsg = (TagapastryMsg)msg;
		if (logger.level <= Logger.FINER)
		{
			if(!(msg instanceof MaintenanceMsg) && !(msg instanceof ReplicationMaintenanceMsg));
				logger.log("Delivered called with " + id + " " + msg);
		}
		
		try {
			//if(tagaMsg instanceof IndexExpansionConfirmationMsg) {
				//System.out.println("Received exp conf. I'm: " + getNodeHandle().getId());
			//}
			
			tagaMsg.execute(this);
		}
		catch(Exception e) {
			e.printStackTrace();
		}
	}
	
	public NodeStatistics getStats() {
		return this.stats;
	}
	
	public String printSize() {
		if(this.indexes != null) {
			//int size = this.getTotalSpaceUsageOfIndexes(this.indexes);
			int index = 0;
			do {
				try {
					this.stats = this.getStatistics(this.indexes);
					return stats.toString();
				}catch(ConcurrentModificationException cce) {
					index++;
				}
			}while(index < 20);
			System.out.println("Puha in printSize on impl");
			
			
			//if(size != 0)
				// Size in bytes, count of indexes,ID
				//return stats.toString();
				//return String.format("%d %d %s", size,this.indexes.size(), node.getId().toString());
			
			//return "";
		} 
		
		return "";
	}

	// To be implemented
	@Override
	public boolean forward(RouteMessage arg0) {
		//System.out.println("Message passed: " + new java.util.Date());
		//if(arg0.getMessage() instanceof IndexExpansionMsg) {
			//System.out.println("Forwarding... dest is: " + arg0.getDestinationId());
			//System.out.println("Forwarding... next hop is: " + arg0.getNextHopHandle().getId());
		//}
		return true;
	}

	/* 
	 * From java doc:
	 * This method is invoked to inform the application that the 
	 * given node has either joined or left the neighbor set of 
	 * the local node, as the set would be returned by the 
	 * neighborSet call.
	 * Experiments shows that its about leaft set as well, 
	 * not only neighborhood sets.
	 * 
	 * (non-Javadoc)
	 * @see rice.p2p.commonapi.Application#update(rice.p2p.commonapi.NodeHandle, boolean)
	 */
	public void update(NodeHandle newNode, boolean joinedOfLeft) {		
		ReplicationStrategy rs = new SimpleReplicationStrategy();
		rs.syncData(this, newNode, joinedOfLeft);
	}

	/*
	 * (non-Javadoc)
	 * @see dk.itu.samd.group2.tagapastry.Tagapastry#tag(java.lang.String, java.lang.String, java.lang.String)
	 * @param tag "news"
	 * @param type "URLS"
	 * @param obj "news.com"
	 */
	@Override
	public boolean tag(String tag, String type, String obj) {
		if (tag.isEmpty() || type.isEmpty() || obj.isEmpty()) {
			throw new IllegalArgumentException("Either the tag, type or object were blank strings");
		}
		Tag newTag = TagFactory.generateTag(tag, type);
		
		if (!taggingBlackBox.shouldTag(newTag, obj)) {
			return false;
		}
		
		TagMsg msg = new TagMsg(newTag, endpoint.getLocalNodeHandle(), obj);
		this.send(msg);
		
		return true;
	}
	
	@Override
	public boolean untag(String tag, String type, String obj) {
		if (tag.isEmpty() || type.isEmpty() || obj.isEmpty()) {
			throw new IllegalArgumentException("Either the tag, type or object were blank strings");
		}
		
		Tag newTag = TagFactory.generateTag(tag, type);
		if (!taggingBlackBox.shouldUntag(newTag, obj)) {
			return false;
		}
		
		UntagMsg msg = new UntagMsg(newTag, endpoint.getLocalNodeHandle(), obj);
		this.send(msg);
		return true;
	}
	

	
	// TODO should be in interface
	public Vector<IndexEntry> getSearchResults(Tag tag) {
		return results.get(tag);
	}

	@Override
	public void search(String tag, String type) {
		if (tag.isEmpty() || type.isEmpty()) {
			throw new IllegalArgumentException("Either the tag or type were blank strings");
		}
		
		String[] searchTags = tag.split(" ");
		for(String s : searchTags) {
			// Use factory!
			// Tag newTag = new Tag(new PastryIdFactory(environment), s, type);
			Tag newTag = TagFactory.generateTag(s, type);
			if(searchTags.length > 1) {
				newTag.makeComplex();
				newTag.setAlternateName(tag);
			}
			
			SearchMsg msg = new SearchMsg(newTag, endpoint.getLocalNodeHandle());
			
			// tmp shall be used only for referencing the result set.
			Tag tmp = TagFactory.duplicateTagAlternateName(newTag);
			Vector<IndexEntry> searchResults = this.results.get(tmp);
			if(searchResults == null) {
				searchResults = new Vector<IndexEntry>();
				this.results.put(tmp,searchResults);
			}

			this.send(msg);
		}
	}

	// TODO should be in interface
	public void putResults(Tag tag, Vector<IndexEntry> newResults) {
		if(tag.isComplex()) {
			tag = TagFactory.duplicateTagAlternateName(tag);
		}
		
		// That will store on "news and sport"
		results.put(tag, newResults);
	}

	public Node getNode() {
		return node;
	}

	public NodeHandle getNodeHandle() {
		return endpoint.getLocalNodeHandle();
	}
	
	@Override
	public Environment getEnvironment() {
		return environment;
	}

	@Override
	public String toString() {
		return instance;
	}

//	public Hashtable<Tag, Index> getIndexes() {
//		return indexes;
//	}

	// due to the internal construction of TagapastryMsg, 
	// either id or nodehandle will be null. 
	public void send(TagapastryMsg msg) {
		if (logger.level <= Logger.FINEST) 
			logger.log("sending " + msg);

		//System.out.println(new java.util.Date() + " sending " + msg.toString());
		//this.client.getMsgCollector().incBytesSent(msg);
		if(!this.client.canSend()) {
			//System.out.println("Blocked for msg type: " + msg.getType());
			return;
		}

		endpoint.route(msg.getRecepientId(), msg, msg.getRecepientNodeHandle());
				
		
		if(shouldCatchData) {
			if(!(msg instanceof MaintenanceMsg) &&
					!(msg instanceof ReplicationMaintenanceMsg)) {
								
				this.client.getMsgCollector().incBytesSent(msg);
				this.client.getMsgCollector().incTotalSent();
			}
		}
		else {
			System.out.println("NOT Catching for.. : " + msg.getClass().getSimpleName());
		}
			
	}

	// TODO should have package visibility
	public TagapastryClient getClient() {
		return client;
	}
	
	// Get leaf set data to update
//	public Hashtable<Id, Hashtable<Tag,Index> > getLeafSetData() {
//		return this.leafsetData;
//	}
	
	/*
	 * Gets leaf set data associated with this node.
	 */
	public Hashtable<NodeHandle, NodeState> getLeafSetData() {
		return this.leafSetData;
	}
	
	/*
	 * Get logger so we can share it among different messages
	 */
	public Logger getLoger() {
		return this.logger;
	}
	
	public void setThreshold(int min, int max)
	{
		if(min == -1 || max == -1 || min >= max) {
			this.retThreshold = 0;
			this.expThreshold = Integer.MAX_VALUE;
			return;
		}
		
		this.retThreshold = min;
		this.expThreshold = max;
	}
	
	public int getRetractionThreshold() {
		return this.retThreshold;
	}
	
	public int getExpansionThreshold() {
		return this.expThreshold;
	}
	
	NodeStatistics getStatistics(Hashtable<Tag,Index> data) {
		NodeStatistics stat = new NodeStatistics(this.getNodeHandle());
		getStatistics(stat, data);
		
		return stat;
	}
	
	synchronized void getStatistics(NodeStatistics ns, Hashtable<Tag,Index> data) {
		int bytes = 0, indexes = 0, entries = 0, children = 0;
		
		int tagSetSize = data.keySet().size();
		Tag[] tags = (Tag[])data.keySet().toArray(new Tag[tagSetSize]);
		
		for(int i=0; i < tags.length; i++) {
			bytes += tags[i].getSize();
		}
		
		int idxSetSize = data.values().size();
		Index[] idexes = (Index[])data.values().toArray(new Index[idxSetSize]);
		for(int i = 0; i < idexes.length; i++) {
			bytes += idexes[i].getIndexSize();
			entries += idexes[i].getEntries().size();
			if(idexes[i].getChildren() != null)
				children += idexes[i].getChildren().size();
		}
		
		ns.setNumberOfBytes(bytes);
		ns.setNumOfIndexes(idexes.length);
		ns.setNumOfEntriesPerNode(entries);
		ns.setNumOfChildren(children);
	}
	
	int getTotalNumOfEntries(Hashtable<Tag, Index> data) {
		if(data == null || data.size() == 0)
			return 0;
		
		int size=0;
		for(Index i : data.values()) {
			size += i.getEntries().size();
		}
		
		return size;
	}
	
	int getTotalSpaceUsageOfIndexes(Hashtable<Tag,Index> data) {
		if(data == null || data.size() == 0)
			return 0;
		
		int space = 0;
		for(Tag tag : data.keySet()) {
			space += tag.getSize();
			Index i = data.get(tag);
			space += i.getIndexSize();
		}
		
		return space;
	}
	
	int getTotalSpaceUsageOfLeafSet() {
		if(leafSetData == null || leafSetData.isEmpty())
			return 0;
		
		int space = 0;
		for(NodeHandle nh : this.leafSetData.keySet()) {
			NodeState data = this.leafSetData.get(nh);
			space += this.getTotalSpaceUsageOfIndexes(data.getIndexes());
		}
		
		return space;
	}
	
	public Index getIndex(Tag tag) {
		return indexes.get(tag);
	}

//	public Iterator<Tag> getIndexKeysIter() {
//		return indexes.keySet().iterator();
//	}

	public void putIndex(Index index) {
		//System.out.println("put index called with: " + index.getTag() + "   empty: " + this.isEmpty() + "   task: " + maintenanceTask);
		indexes.put(index.getTag(), index);
		if (this.maintenanceTask == null ) {
			//System.out.println("starting main task");
			maintenanceTask = endpoint.scheduleMessage(
					new MaintenanceMsg( this.getNodeHandle() ), 4000, 4000);
		}
		
		// No need to replicate indexes, if you don't have them so moving procedure here.
		// Start after 4 seconds, every 20
		if(replicationMntTask == null) {
			this.replicationMntTask = endpoint.scheduleMessage(new ReplicationMaintenanceMsg(this.getNodeHandle()), 4000,20000);
		}
	}

	public void checkToStopMaintanceTask() {
		System.out.println("maintenanceTask: " + (maintenanceTask != null));
		if (this.isEmpty() && maintenanceTask != null) {
			System.out.println("canceling task");
			maintenanceTask.cancel();
			maintenanceTask = null;
		}
	}

	private boolean isEmpty() {
		if (indexes.isEmpty()) {
			return true;
		}
		for (Index index : this) {
			boolean empty = index.isEmpty();
			if (!empty) {
				return false;
			}
		}
		return true;
	}
	
	public boolean hasIndexes() {
		return !indexes.isEmpty();
	}

	@Override
	public Iterator<Index> iterator() {
		return indexes.values().iterator();
	}
	
	public Hashtable<Tag, Index> getIndexes() {
		return indexes;
	}

	public void removeIndex(Tag tag) {
		indexes.remove(tag);
		checkToStopMaintanceTask();
	}
	
	public void catchData() {
		this.shouldCatchData = true;
	}
	
	/*
	 * Make sure to remove after testing
	 */
	public void flushIndexes() {
		resetIndexes(this.indexes);
		// Clear hash table for new set of tests
		this.indexes.clear();
		this.indexes = null;
		this.indexes = new Hashtable<Tag, Index>();
		
		for(NodeState ns : this.leafSetData.values()) {
			this.resetIndexes(ns.getIndexes());
		}
		
		// Clear also leaf set information
		this.leafSetData.clear();
		this.leafSetData = null;
		this.leafSetData = new Hashtable<NodeHandle, NodeState>();
	}
	
	private void resetIndexes(Hashtable<Tag,Index> idx) {
		Set<Tag> set = idx.keySet();
		Tag[] tags = (Tag[])set.toArray(new Tag[set.size()]);
		tags = null;
		
		Collection<Index> seti = idx.values();
		Index[] aindexes = (Index[])seti.toArray(new Index[seti.size()]);
	
		for(int i=0; i < aindexes.length; i++) {
			Vector<IndexRelative> rel = aindexes[i].getChildren();
			if(rel == null) continue;
			
//			for(int j=0; j < rel.size(); j++) {
//				IndexRelative r = rel.get(j);
//				for(int k = 0; k < r.getMsgBuffer().size(); k++) 
//				{
//					client.getMsgCollector().incBytesSent(r.getMsgBuffer().get(k));
//				}
//			}
			
			// Null relatives
			rel = null;
		}
		
		aindexes = null;
	}
	
	public void processBuffer() {

		Collection<Index> seti = indexes.values();
		Index[] aindexes = (Index[])seti.toArray(new Index[seti.size()]);
	
		for(int i=0; i < aindexes.length; i++) {
			Vector<IndexRelative> rel = aindexes[i].getChildren();
			if(rel == null) continue;
			
			for(int j=0; j < rel.size(); j++) {
				IndexRelative r = rel.get(j);
				for(int k = 0; k < r.getMsgBuffer().size(); k++) 
				{
					client.getMsgCollector().incBytesSent(r.getMsgBuffer().get(k));
				}
			}
			
			// Null relatives
			//rel = null;
		}
	}
}
